package com.nx.platform.es.biz.esspider.handler;

import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.primitives.Ints;
import com.google.common.primitives.Longs;
import com.nx.platform.es.biz.esspider.resource.DBManager;
import com.nx.platform.es.common.utils.MoreMaps;
import com.nx.platform.es.common.utils.MoreSplitters;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.math.NumberUtils;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * 提供一对多处理能力：sum，max，min，avg，multi
 *
 * @author wangjianying
 * @since Aug 19, 2019
 */
@HandlerDefine(HandlerType.DB)
public abstract class AbstractDBGroupByHandler extends AbstractDBHandler {

    protected String sql;
    protected String idField;
    protected Map<GroupType, Set<String>> fields = new HashMap<>();
    protected boolean toString;

    public AbstractDBGroupByHandler(String identity, DBManager dbManager) {
        super(identity, dbManager);
    }

    @Override
    public void init(Map<?, ?> settings) throws Exception {
        sql = MoreMaps.getString(settings, "sql");
        idField = MoreMaps.getString(settings, "idField");
        Set<String> sumField = MoreMaps.getStringSet(settings, "sumField", MoreSplitters.COMMA);
        Set<String> minField = MoreMaps.getStringSet(settings, "minField", MoreSplitters.COMMA);
        Set<String> maxField = MoreMaps.getStringSet(settings, "maxField", MoreSplitters.COMMA);
        Set<String> multiField = MoreMaps.getStringSet(settings, "multiField", MoreSplitters.COMMA);
        toString = MoreMaps.getBooleanValue(settings, "toString", false);
        if (CollectionUtils.isNotEmpty(sumField)) {
            fields.putIfAbsent(GroupType.SUM, sumField);
        }
        if (CollectionUtils.isNotEmpty(minField)) {
            fields.putIfAbsent(GroupType.MIN, minField);
        }
        if (CollectionUtils.isNotEmpty(maxField)) {
            fields.putIfAbsent(GroupType.MAX, maxField);
        }
        if (CollectionUtils.isNotEmpty(multiField)) {
            fields.putIfAbsent(GroupType.MULTI, multiField);
        }
        Preconditions.checkNotNull(getJdbcTemplate());
        Preconditions.checkArgument(!Strings.isNullOrEmpty(sql), "sql null or empty");
        Preconditions.checkArgument(!Strings.isNullOrEmpty(idField), "idField null or empty");
    }

    protected enum GroupType {
        SUM {
            @Override
            Object parse(Collection<Object> values) throws NumberFormatException {
                checkIsNumber(values);
                Object first = values.iterator().next();
                if (first instanceof Integer) {
                    return values.stream().reduce((i1, i2) -> (int) i1 + (int) i2).orElse(0);
                } else if (first instanceof Long) {
                    return values.stream().reduce((i1, i2) -> (long) i1 + (long) i2).orElse(0);
                }
                return 0;
            }
        },
        MAX {
            @Override
            Object parse(Collection<Object> values) throws NumberFormatException {
                checkIsNumber(values);
                Object first = values.iterator().next();
                if (first instanceof Integer) {
                    return values.stream().reduce((i1, i2) -> Ints.max((int) i1 + (int) i2)).orElse(0);
                } else if (first instanceof Long) {
                    return values.stream().reduce((i1, i2) -> Longs.max((long) i1 + (long) i2)).orElse(0);
                }
                return 0;
            }
        },
        MIN {
            @Override
            Object parse(Collection<Object> values) throws NumberFormatException {
                checkIsNumber(values);
                Object first = values.iterator().next();
                if (first instanceof Integer) {
                    return values.stream().reduce((i1, i2) -> Ints.min((int) i1 + (int) i2)).orElse(0);
                } else if (first instanceof Long) {
                    return values.stream().reduce((i1, i2) -> Longs.min((long) i1 + (long) i2)).orElse(0);
                }
                return 0;
            }
        },
        MULTI {
            @Override
            Object parse(Collection<Object> values) throws NumberFormatException {
                return values;
            }
        },
        AVG {
            @Override
            Object parse(Collection<Object> values) throws NumberFormatException {
                checkIsNumber(values);
                Object first = values.iterator().next();
                if (first instanceof Integer) {
                    return (int) values.stream().reduce((i1, i2) -> (int) i1 + (int) i2).orElse(0) / values.size();
                } else if (first instanceof Long) {
                    return (long) values.stream().reduce((i1, i2) -> (long) i1 + (long) i2).orElse(0) / values.size();
                }
                return 0;
            }
        };

        abstract Object parse(Collection<Object> values) throws NumberFormatException;

        static void checkIsNumber(Collection<Object> values) throws NumberFormatException {
            try {
                Preconditions.checkNotNull(values);
                values.forEach(value -> Preconditions.checkArgument(NumberUtils.isNumber(value.toString())));
            } catch (Exception e) {
                throw new NumberFormatException(e.getMessage() + "values type illegal");
            }

        }
    }
}